#include "shm.h"
#include "cJSON.h"
#include "common.h"
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include "yuan_pthread.h"
#include "msg_handler.h"
#include "MQTTClient.h" // MQTT客户端函数

MQTTClient yuan_pub_client = NULL;    // MQTT发布客户端
MQTTClient yuan_sub_client = NULL;    // MQTT订阅客户端
void* shm_addr =NULL;    // 共享内存地址
int yuan_quit = 0;                  // 程序退出标志



void* yuan_shm_read_thread(void* arg) {
    devdata data;
    char* json_buf = NULL;  // 用于存储JSON序列化结果

    while (!yuan_quit) {
        if (shm_read(shm_addr, &data) != 0) {
            printf("[read_shm_thread] 共享内存读取失败,1秒后重试\n");
            sleep(1);
            continue;
        }

        // 2. 显式“反序列化”：提取结构体字段
        float tem = data.tem;
        float hum = data.hum;
        printf("[反序列化] 解析结果:tem=%.2f, hum=%.2f\n",tem,hum);

        // 3. 重新序列化为JSON
        free(json_buf);  // 释放上一次的JSON内存
        json_buf = json_in(&data);  // 传入结构体，生成JSON
        printf("[序列化]结果:%s\n",json_buf);
        if (!json_buf) {
            printf("[共享内存线程] JSON序列化失败");
            sleep(1);
            continue;
        }

        // 4. 发送MQTT消息（JSON格式）
       if (yuan_mqtt_publish(yuan_pub_client, json_buf) != 0) {
           printf("MQTT发布失败\n");
       } 
       else {
           printf("MQTT发布成功:%s\n",json_buf);
       }
       sleep(1);
      }

    printf("[read_shm_thread] 退出\n");
    return NULL;
    

    free(json_buf);  // 退出前释放内存
    printf("[共享内存线程] 退出\n");
    return NULL;

}

void* yuan_mqtt_public_thread(void* arg) {
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;

    if ((rc = MQTTClient_create(&yuan_pub_client, YUAN_ADDRESS, YUAN_PUB_CLIENT,
        MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS) {
        printf("[mqtt_publish_thread] 创建客户端失败，错误码: %d\n", rc);
        yuan_quit = 1;
        return NULL;
    }

    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    if ((rc = MQTTClient_connect(yuan_pub_client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("[mqtt_publish_thread] 连接失败，错误码: %d\n", rc);
        MQTTClient_destroy(&yuan_pub_client);
        yuan_quit = 1;
        return NULL;
    }
    printf("[mqtt_publish_thread] 发布客户端连接成功（主题：%s)\n", YUAN_PUB_TOPIC);

    while (!yuan_quit) sleep(1);

    MQTTClient_disconnect(yuan_pub_client, YUAN_TIMEOUT);
    MQTTClient_destroy(&yuan_pub_client);
    printf("[mqtt_publish_thread] 退出\n");
    return NULL;
}

void* yuan_mqtt_subscribe_thread(void* arg)
{
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;

    if ((rc = MQTTClient_create(&yuan_sub_client, YUAN_ADDRESS, YUAN_SUB_CLIENT,
        MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS) {
        printf("[mqtt_subscribe_thread] 创建客户端失败，错误码: %d\n", rc);
        yuan_quit = 1;
        return NULL;
    }

    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    MQTTClient_setCallbacks(yuan_sub_client, NULL, yuan_connlost, yuan_msgarrvd, NULL);

    if ((rc = MQTTClient_connect(yuan_sub_client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("[mqtt_subscribe_thread] 连接失败，错误码: %d\n", rc);
        MQTTClient_destroy(&yuan_sub_client);
        yuan_quit = 1;
        return NULL;
    }
    if ((rc = MQTTClient_subscribe(yuan_sub_client, YUAN_SUB_TOPIC, YUAN_QOS)) != MQTTCLIENT_SUCCESS) {
        printf("[mqtt_subscribe_thread] 订阅失败，错误码: %d\n", rc);
        MQTTClient_disconnect(yuan_sub_client, YUAN_TIMEOUT);
        MQTTClient_destroy(&yuan_sub_client);
        yuan_quit = 1;
        return NULL;
    }
    printf("[mqtt_subscribe_thread] 订阅客户端连接成功（主题：%s)\n", YUAN_SUB_TOPIC);

    while (!yuan_quit) sleep(1);

    MQTTClient_unsubscribe(yuan_sub_client, YUAN_SUB_TOPIC);
    MQTTClient_disconnect(yuan_sub_client, YUAN_TIMEOUT);
    MQTTClient_destroy(&yuan_sub_client);
    printf("[mqtt_subscribe_thread] 退出\n");
    return NULL;
}


int create_yuan_threads(pthread_t* t1, pthread_t* t2,pthread_t* t3)
{
    if (pthread_create(t1, NULL, yuan_shm_read_thread, shm_addr) != 0) {
        printf("[create_all_threads] 创建read_shm_thread失败\n");
        return -1;
    }
    if (pthread_create(t2, NULL, yuan_mqtt_public_thread, NULL) != 0) {
        printf("[create_all_threads] 创建mqtt_publish_thread失败\n");
        return -1;
    }
    if (pthread_create(t3, NULL, yuan_mqtt_subscribe_thread, NULL) != 0) {
        printf("[create_all_threads] 创建mqtt_publish_thread失败\n");
        return -1;
    }

    return 0;
}